package com.yk.mr.dao.application;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobStatus;
import org.apache.hadoop.mapred.TaskReport;
import org.apache.hadoop.util.Tool;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.net.URL;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;
import java.util.regex.Pattern;


public class JobMonitor extends Configured implements Tool {

    public static final String HOST="namenode01";//主机名称
    public static final int JOBTRACKER_PORT=8088;//jobtracker rpc 端口
    public static final String NAMENODE_PORT="hdfs://namenode01:9000";//hdfs地址

    public static void main(String[] args) throws IOException, ParseException {
        System.out.println("The name of task:");
        InetSocketAddress jobTracker = new InetSocketAddress(HOST,JOBTRACKER_PORT);//远程连接jobTracker
        JobClient jobClient = new JobClient(jobTracker,getsConf());//初始化jobClient
        JobStatus[] jobStatusAll = jobClient.getAllJobs();
        JobStatus jobStatus = null;
        if (jobStatusAll==null||jobStatusAll.length<=0){
            //当前集群中没有job,集群重启或者原本就没有job都有可能造成这样的结果
            ClusterStatus clusterStatus = jobClient.getClusterStatus(true);
            System.out.println("集群"+clusterStatus+"当天到目前为止还没有运行过job!");
        }else{
            System.out.println("job的总数是："+jobStatusAll.length);
            for (int i=0;i<jobStatusAll.length;i++){
                //输出job列表中所有job的信息
                jobStatus = jobStatusAll[i];
                String jobName = jobClient.getJob(jobStatus.getJobID()).getJobName();
                int jobState = jobStatus.getRunState();
                String state = "";
                switch(jobState){
                    case 1:
                        state="RUNNING";
                        break;
                    case 2:
                        state="SUCCEEDED";
                        break;
                    case 3:
                        state = "FAILED";
                        break;
                    case 4:
                        state ="PREP";
                        break;
                    case 5:
                        state = "KILLED";
                        default:
                            break;
                }

                System.out.println("job Id:"+jobStatus.getJobID());
                System.out.println("job Name:"+jobName);
                System.out.println("job User:"+jobStatus.getUsername());
                System.out.println("job Map-progress : " + jobStatus.mapProgress());
                System.out.println("job Map-total : "+jobClient.getMapTaskReports(jobStatus.getJobID()).length);
                System.out.println("job Reduce-progress : " + jobStatus.reduceProgress());
                System.out.println("job Reduce-total : "+jobClient.getReduceTaskReports(jobStatus.getJobID()).length);
                System.out.println("job Cleanup-progress : " + jobStatus.cleanupProgress());
                System.out.println("job Setup-progress : " + jobStatus.setupProgress());
                System.out.println("job Priority : " + jobStatus.getJobPriority());
                System.out.println("job Counters " + jobClient.getJob(jobStatus.getJobID()).getCounters());
                System.out.println("job Start-time:"+jobStatus.getStartTime());

                //通过jobtracker网页连接获取其他信息
                URL url = new URL(jobClient.getJob(jobStatus.getJobID()).getTrackingURL());
                BufferedReader br= new BufferedReader(new InputStreamReader(url.openStream()));
                String w="";
                StringBuffer sb=new StringBuffer("");
                while((w=br.readLine())!= null){
                    sb.append(w);
                }
                br.close();//上边一小段是获取jobtracker网页信息
                System.out.println("job Finish-time:"+jobStatus.getFinishTime());//job结束时间， 因为api中 jobStatus.getFinishTime()函数找不到，所以自己写了一个。

                Pattern p = Pattern.compile("Finished at:</b>.*?<br>");//过滤获取的信息，得到job所有时间
                java.util.regex.Matcher rm = p.matcher(sb);
                while (rm.find()){
                    String tmp =rm.group();
                    String str = tmp.substring(17,tmp.length()-4);
                    Date date = parse(str, "EEE MMM dd HH:mm:ss zzz yyyy", Locale.US);
                    SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    String f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
                    Date date1 = format.parse(f);
                    System.out.println("job Finish-time:"+date1.getTime());//job结束时间， 因为api中 jobStatus.getFinishTime()函数找不到，所以自己写了一个。
                }

                Pattern p1 = Pattern.compile("Finished in:</b>.*?<br>");//过滤获取的信息，得到job所有时间
                java.util.regex.Matcher  rm1 = p1.matcher(sb);
                while(rm1.find()){
                    String tmp = rm1.group();
                    System.out.println("job Used-time:"+tmp.substring(17, tmp.length()-4));//总共耗时
                }
                System.out.println("job State:"+state);//状态

                System.out.println("Map task:");//map详情信息
                TaskReport[] mtp = jobClient.getMapTaskReports(jobStatus.getJobID());
                for (int j=0;j<mtp.length;j++){
                    System.out.println("task id:"+mtp[j].getTaskId());
                    System.out.println("task Start-time:"+mtp[j].getStartTime());
                    System.out.println("task Finish-time:"+mtp[j].getFinishTime());
                    System.out.println("task Progress:"+mtp[j].getProgress());
                    System.out.println("task Status:"+mtp[j].getCurrentStatus());
                    System.out.println("task Counters:"+mtp[j].getCounters());
                }

                System.out.println("Reduce task:");//reduce详情信息
                TaskReport[] rtp = jobClient.getReduceTaskReports(jobStatus.getJobID());
                for(int j=0;j<rtp.length;j++){
                    System.out.println("task id:"+rtp[j].getTaskId());
                    System.out.println("task Start-time:"+rtp[j].getStartTime());
                    System.out.println("task Finish-time:"+rtp[j].getFinishTime());
                    System.out.println("task Progress:"+rtp[j].getProgress());
                    System.out.println("task Status:"+rtp[j].getCurrentStatus());
                    System.out.println("task Counters:"+rtp[j].getCounters());
                }

            }
        }
    }

    public static Date parse(String str,String pattern,Locale locale){
        //转为本地的时间格式
        if(str==null || pattern ==null){
            return null;
        }
        try{
            return new SimpleDateFormat(pattern,locale).parse(str);
        }catch (ParseException e){
            e.printStackTrace();
        }
        return null;
    }

    public static Configuration getsConf(){
        //设置配置信息
        Configuration conf =new Configuration();
//        conf.set("mapred.job.tracker",HOST+":"+JOBTRACKER_PORT);
//        conf.set("fs.default.name",NAMENODE_PORT);
        conf.set("mapreduce.jobtracker.address",HOST+":"+JOBTRACKER_PORT);
        conf.set("fs.defaultFS",NAMENODE_PORT);
        return conf;
    }
    @Override
    public int run(String[] strings) throws Exception {
        return 0;
    }
}
